package com.ruoyi.framework.mq.RabbitMQ;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


import java.util.HashMap;
import java.util.Map;

/**
 * rabbitMq配置类
 */
@Configuration
public class RabbitMqConfig {
    private static final Logger log=LoggerFactory.getLogger(RabbitMqConfig.class);

    //存在多个交换机和多个队列 需要在 @Bean(name = "")中加上名称区分 否则会报错


    //交换机 类型
    //routing_key==路由键
    //Direct：direct 类型的行为是"先匹配, 再投送". 即在绑定时设定一个 routing_key, 消息的routing_key 匹配时, 才会被交换器投送到绑定的队列中去.
    //Direct<直接>：1对1-----一个消息只能被一个消费者消费
    //Topic：按规则转发消息（最灵活） 1对多-----一个消息可以被多个消费者消费
    //Fanout：转发消息到所有绑定队列 广播
    //Headers：设置 header attribute 参数类型的交换机

/********************************************************direct队列******************************************************************************/
    /**
     * direct队列
     * @return
     */
    @Bean
    public Queue directQueue(){
        Map<String, Object> arguments = new HashMap<>(4);
        /**
         * Queue(String name, boolean durable, boolean exclusive, boolean autoDelete，arguments)
         * name 队列名称
         * durable 是否持久化队列
         * exclusive 是否为独占队列（只允许单个消费者）
         * autoDelete 是否自定删除队列，（当没有消费者者时自动删除）
         * arguments 申明参数
         */
        Queue queue = new Queue(RabbitMqConstants.DIRECT_QUEUE,true,false,false,arguments);
        return queue;
    }

    /**
     * direct模式交换机  适用场景 异步通知  比如 商品订单下单成功 异步通知到库存系统减库存
     * 交换机的类型为 direct
     * direct 类型的行为是"先匹配, 再投送". 即在绑定时设定一个 routing_key, 消息的routing_key 匹配时, 才会被交换器投送到绑定的队列中去.
     * @return 返回交换机 注入spring中
     */
    @Bean
    public Exchange directExchange() {
        Map<String, Object> args = new HashMap<>();
        /**
         *AbstractExchange(String name, boolean durable, boolean autoDelete, Map<String, Object> arguments)
         * name 交换器名称
         * durable 是否持久化交换器
         * autoDelete 是否自动删除
         * args 申明参数
         */
        Exchange exchange = new DirectExchange(RabbitMqConstants.DIRECT_EXCHANGE, true, false,args);
        return exchange;
    }

    /**
     * 将directQueue队列绑定到 direct交换机上 指明路由键
     * @return
     */
    @Bean
    public Binding directBinding() {
        //bind 绑定队列
        //to 到哪个交换机上
        //with 路由键 表明是哪个路由的
        return BindingBuilder.bind(directQueue()).to(directExchange()).with(RabbitMqConstants.DIRECT_ROUTINGKEY).noargs();
    }
/********************************************************direct队列完******************************************************************************/
/********************************************************fanout队列开始******************************************************************************/
    /**
     * fanout队列1
     * @return
     */
    @Bean(name = "fanoutQueue1")
    public Queue fanoutQueue1(){
        //参数
        // String name 队列名称
        // boolean durable 启的时候是否保留这个队列 true保留 false不保留
        Queue queue = new Queue(RabbitMqConstants.FANOUT_QUEUE_1);
        return queue;
    }

    /**
     * fanout队列2
     * @return
     */
    @Bean(name = "fanoutQueue2")
    public Queue fanoutQueue2(){
        Queue queue = new Queue(RabbitMqConstants.FANOUT_QUEUE_2);
        return queue;
    }

    /**
     * 订阅交换机 适用场景 如 群发信息
     * @return
     */
    @Bean
        FanoutExchange fanoutExchange() {
        //交换机构造参数 String name, boolean durable, boolean autoDelete
        //String name 交换机名称
        //boolean durable 重启的时候是否保留这个交换机 true保留 false不保留
        //boolean autoDelete 交换机中没有数据是否删除这个交换机 true删除 false不删除
      return new FanoutExchange(RabbitMqConstants.FANOUT_EXCHANGE);
    }

    /**
     * 将fanout队列1 绑定到 fanoutExchange交换机上
     * @return
     */
    @Bean
    public Binding fanoutBinding1(@Qualifier("fanoutQueue1")Queue fanoutQueue1,FanoutExchange fanoutExchange ) {
        //bind 绑定队列
        //to 到哪个交换机上
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }

    /**
     * 将fanout队列2 绑定到 fanoutExchange交换机上
     * @return
     */
    @Bean
    public Binding fanoutBinding2(@Qualifier("fanoutQueue2")Queue fanoutQueue2,FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }


/********************************************************fanout队列完******************************************************************************/
/********************************************************topic队列开始******************************************************************************/
    /**
     * topic队列1
     * @return
     */
    @Bean
    public Queue topicQueue1(){
        Queue queue = new Queue(RabbitMqConstants.TOPIC_QUEUE_1,true);
        return queue;
    }

    /**
     * topic队列2
     * @return
     */
    @Bean
    public Queue topicQueue2(){
        Queue queue = new Queue(RabbitMqConstants.TOPIC_QUEUE_2,true);
        return queue;
    }
    /**
     * topicExchange topic交换机
     * @return
     */
    @Bean
    public Exchange topicExchange() {
        Exchange exchange = new TopicExchange(RabbitMqConstants.TOPIC_EXCHANGE,true,false);
        return exchange;
    }

    /**
     * 将topic队列1 绑定到 topicExchange交换机上
     * topic.# .# 代表任意个单词
     * @return
     */
    @Bean
    public Binding topicBinding1() {
        return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("topic.#").noargs();
    }

    /**
     * 将topic队列2 绑定到 topicExchange交换机上
     * topic.* * 仅代表一个单词
     * @return
     */
    @Bean
    public Binding topicBinding2() {
        return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("topic.*").noargs();
    }
/********************************************************topic队列完*****************************************************************************/
/********************************************************TTL延迟队列队列开始*****************************************************************************/
    // 创建一个立即消费队列
    @Bean(QueueName.ImmediateQueue)
    public Queue immediateQueue() {
        return new Queue(QueueName.ImmediateQueue);
    }
    // 创建一个立即消费交换机
    @Bean(ExchangeName.IMMEDIATE)
    public DirectExchange immediateExchange() {
        return new DirectExchange(ExchangeName.IMMEDIATE);
    }

    // 把 立即消费的队列 和 立即消费的exchange 绑定在一起
    @Bean
    public Binding immediateBinding(@Qualifier(QueueName.ImmediateQueue) Queue queue, @Qualifier(ExchangeName.IMMEDIATE) DirectExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(RoutingKey.IMMEDIATE_ROUTING_KEY);
    }


    // 创建一个延时队列
    @Bean(QueueName.DelayQueue)
    public Queue delayQueue() {
        Map<String, Object> params = new HashMap<>();

        // x-dead-letter-exchange 声明了队列里的死信转发到的DLX名称，
        params.put("x-dead-letter-exchange", ExchangeName.IMMEDIATE);

        // x-dead-letter-routing-key 声明了这些死信在转发时携带的 routing-key
        params.put("x-dead-letter-routing-key", RoutingKey.IMMEDIATE_ROUTING_KEY);

        // 设置队列中消息的过期时间，单位 毫秒
        params.put("x-message-ttl", 5 * 1000);

        return new Queue(QueueName.DelayQueue, true, false, false, params);
    }
    // 创建一个延时交换机
    @Bean(ExchangeName.DELAY)
    public DirectExchange deadLetterExchange() {
        return new DirectExchange(ExchangeName.DELAY);
    }

    // 把 延迟消费的队列 和 延迟消费的exchange 绑定在一起
    @Bean
    public Binding delayBinding(@Qualifier(QueueName.DelayQueue) Queue queue, @Qualifier(ExchangeName.DELAY) DirectExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(RoutingKey.DELAY_KEY);
    }
    // 常量
    public interface ExchangeName {
        String IMMEDIATE = "immediate";
        String DELAY = "delay";
    }
    public interface QueueName {
        String ImmediateQueue = "ImmediateQueue";
        String DelayQueue = "DelayQueue";
        String XImmediateQueue = "XImmediateQueue";
    }
    public interface RoutingKey {
        String IMMEDIATE_ROUTING_KEY = "immediate.key";
        String DELAY_KEY = "delay.key";
        String XDELAY_KEY = "xdelay.key";
    }
    /********************************************************TTL延迟队列队列完*****************************************************************************/
/********************************************************插件延迟队列队列开始*****************************************************************************/
    // 创建一个立即消费队列
    @Bean(QueueName.XImmediateQueue)
    public Queue immediateXQueue() {
        // 第一个参数是创建的queue的名字，第二个参数是是否支持持久化
        return new Queue(QueueName.XImmediateQueue, true);
    }

    @Bean
    public CustomExchange delayExchange() {
        Map<String, Object> args = new HashMap<String, Object>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange(RabbitMqConstants.DELAYED_EXCHANGE, "x-delayed-message", true, false, args);
    }

    @Bean
    public Binding bindingNotify() {
        return BindingBuilder.bind(immediateXQueue()).to(delayExchange()).with(RoutingKey.XDELAY_KEY).noargs();
    }
    /********************************************************插件延迟队列队列完*****************************************************************************/
    /********************************************************消息确认机制开始*****************************************************************************/
  /**
      * 定制化amqp模版
      *
      * ConfirmCallback接口用于实现消息发送到RabbitMQ交换器后接收ack回调   即消息发送到exchange  ack
      * ReturnCallback接口用于实现消息发送到RabbitMQ 交换器，但无相应队列与交换器绑定时的回调  即消息发送不到任何一个队列中  ack
      */
    @Bean
    public RabbitTemplate getTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate template = new RabbitTemplate(connectionFactory);

        // 消息发送成功到交换器Exchange后触发回调
        template.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                //  可以进行消息入库操作
                log.info("消息唯一标识 correlationData = {}", correlationData.getId());
                log.info("确认结果 ack = {}", ack);
                log.info("失败原因 cause = {}", cause);
            }
        });

        // 配置这个，下面的ReturnCallback 才会起作用
        template.setMandatory(true);
        // 如果消息从交换器发送到对应队列失败时触发（比如 根据发送消息时指定的routingKey找不到队列时会触发）
        template.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                //  可以进行消息入库操作
                log.info("消息主体 message = {}", message);
                log.info("回复码 replyCode = {}", replyCode);
                log.info("回复描述 replyText = {}", replyText);
                log.info("交换机名字 exchange = {}", exchange);
                log.info("路由键 routingKey = {}", routingKey);
            }
        });
        return template;
    }
}


